package com.qupeng.demo.kafka.kafkaapache.security;

import kafka.security.authorizer.AclAuthorizer;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.server.authorizer.Action;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import org.apache.kafka.server.authorizer.AuthorizationResult;

import java.util.Collections;
import java.util.List;
import java.util.Set;

import static org.apache.kafka.common.protocol.ApiKeys.CREATE_ACLS;
import static org.apache.kafka.common.protocol.ApiKeys.DELETE_ACLS;
import static org.apache.kafka.server.authorizer.AuthorizationResult.DENIED;

public class CustomAuthorizer extends AclAuthorizer {
    private static final Set<Short> internalOps = Utils.mkSet(CREATE_ACLS.id, DELETE_ACLS.id);
    private static final String internalListener = "INTERNAL";

    @Override
    public List<AuthorizationResult> authorize(AuthorizableRequestContext context, List<Action> actions) {
        // 如果不是内部请求，并且ACL是创建和删除的操作，禁止执行
        if (!context.listenerName().equals(internalListener) &&
                internalOps.contains((short) context.requestType())) {
            return Collections.nCopies(actions.size(), DENIED);
        } else {
            // 如果是内部请求，执行默认的授权逻辑
            return super.authorize(context, actions);
        }
    }
}